package com.geniusk.mqtt.strategy.impl;

import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.geniusk.common.utils.RedisUtil;
import com.geniusk.common.utils.SpringContextUtil;
import com.geniusk.mqtt.strategy.MessageStrategy;
import com.geniusk.mqtt.utils.MqttTopicUtil;
import com.geniusk.mqtt.utils.MqttUtil;
import com.geniusk.system.model.pojo.ModelProperty;
import com.geniusk.system.service.ModelPropertyService;
import com.geniusk.wss.core.BrokerConstants;
import com.geniusk.wss.core.TopicConstants;
import com.geniusk.wss.utils.WsUtil;
import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Set;

/**
 * publish Ack消息
 */
@Slf4j
@Component
public class PublishAckMessageStrategy  implements MessageStrategy {

    private final RedisUtil redisUtil = SpringContextUtil.getBean(RedisUtil.class);

    private final RabbitTemplate rabbitTemplate = SpringContextUtil.getBean(RabbitTemplate.class);

    private final ModelPropertyService modelPropertyService =  (ModelPropertyService) SpringContextUtil.getBean("modelPropertyServiceImpl");

    @Override
    public void sendResponseMessage(Channel channel, MqttMessage mqttMessage) {
        MqttPublishMessage mqttPublishMessage = (MqttPublishMessage)mqttMessage;

        //解析MqttQoS
        MqttFixedHeader mqttFixedHeader = mqttPublishMessage.fixedHeader();
        MqttQoS mqttQoS = mqttFixedHeader.qosLevel();

        int packageId = mqttPublishMessage.variableHeader().packetId();
        String topicName = mqttPublishMessage.variableHeader().topicName();
        //解析消息
        byte[] headBytes = new byte[mqttPublishMessage.payload().readableBytes()];
        mqttPublishMessage.payload().readBytes(headBytes);
        String data = new String(headBytes);
        JSONObject json = JSONUtil.parseObj(data);
        Long deviceId = Long.parseLong(MqttTopicUtil.getDeviceId(topicName));
        String topicType = MqttTopicUtil.getTopicType(topicName);
        if(topicType.equals(TopicConstants.PROPERTY_TOPIC) && !MqttTopicUtil.hasReply(topicName)){
            List<ModelProperty> modelPropertyList = modelPropertyService.selectModelPropertyByDeviceId(deviceId);
            for(ModelProperty modelProperty : modelPropertyList) {
                String field = modelProperty.getField();
                if(json.containsKey(field)){
                    modelProperty.setValue(json.get(field).toString());
                }
            }
            String jsonStr = JSONUtil.toJsonStr(modelPropertyList);
            // 发送消息到RabbitMQ
//            rabbitTemplate.convertAndSend(MQConfig.MQTT_EXCHANGE,MQConfig.ROUTING_KEY_MQTT_SEND + deviceId, jsonStr);
            // 发送消息到用户
            Set<Long> identityIdList = redisUtil.getCacheObject(BrokerConstants.DEVICE_SUB_IDENTITY + deviceId);
            if(identityIdList != null){
                for(Long identityId : identityIdList){
                    Set<Long> deviceIdList = redisUtil.getCacheObject(BrokerConstants.IDENTITY_SUB_DEVICE + identityId);
                    if(deviceIdList != null){
                        if(deviceIdList.contains(deviceId)){
                            WsUtil.sendToWs(identityId, data);
                        }
                    }
                }
            }
            // 转发到MQTT客户端
            MqttUtil.sendToMqtt(topicName, data);
        }
        switch (mqttQoS) {
            case AT_MOST_ONCE:
                //最多一次
                break;
            case AT_LEAST_ONCE:
                //至少一次
                processAtLeastOnceMsg(channel,mqttFixedHeader,packageId);
                break;
            case EXACTLY_ONCE:
                //刚好一次
                processExactlyOnceMsg(channel,mqttFixedHeader,packageId);
                break;
            default:
                break;
        }
    }

    public void processAtLeastOnceMsg(Channel channel,MqttFixedHeader mqttFixedHeader,int packageId){
        MqttMessageIdVariableHeader mqttMessageIdVariableHeaderBack = MqttMessageIdVariableHeader.from(packageId);
        MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.PUBACK,mqttFixedHeader.isDup(), MqttQoS.AT_MOST_ONCE, mqttFixedHeader.isRetain(), 0x02);
        MqttPubAckMessage pubAckMsg = new MqttPubAckMessage(mqttFixedHeaderBack, mqttMessageIdVariableHeaderBack);
        channel.writeAndFlush(pubAckMsg);
    }

    public void processExactlyOnceMsg(Channel channel,MqttFixedHeader mqttFixedHeader,int packageId){
        MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.PUBREC,false, MqttQoS.AT_LEAST_ONCE,false,0x02);
        MqttMessageIdVariableHeader mqttMessageIdVariableHeaderBack = MqttMessageIdVariableHeader.from(packageId);
        MqttMessage pubAckMsg = new MqttMessage(mqttFixedHeaderBack,mqttMessageIdVariableHeaderBack);
        channel.writeAndFlush(pubAckMsg);
    }

}
